package com.audaque.springboot.foshanupload.kafka.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

/**
 * @author zgb
 * @desc ...
 * @date 2023-05-23 21:00:40
 */
@Configuration
public class KafkaConsumerDiyConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String kafkaServer;

    private Map<String, Object> consumerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // offset偏移量规则设置：
        // (1)、earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
        // (2)、latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
        // (3)、none：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        return props;
    }


    /**
     * 消费工厂公共配置
     * @param factory
     */
    private void commonFactoryAssignment(ConcurrentKafkaListenerContainerFactory<Integer, String> factory) {
        // 拉取超时时间
        factory.getContainerProperties().setPollTimeout(3000);
        // 批次监听
        factory.setBatchListener(true);
        // 批次提交
        factory.getContainerProperties().setAckMode((ContainerProperties.AckMode.MANUAL));
    }



    @Bean("ackContainerFactory")
    public ConcurrentKafkaListenerContainerFactory ackContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        Map<String, Object> propsMap = consumerProps();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory(propsMap));
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }





    @Bean("batchContainerFactory")
    public ConcurrentKafkaListenerContainerFactory listenerContainer() {
        ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();


        Map<String, Object> propsMap = consumerProps();

        //一次拉取消息数量
        propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");

        container.setConsumerFactory(new DefaultKafkaConsumerFactory(propsMap));
        //设置并发量，小于或等于Topic的分区数
        container.setConcurrency(5);
        //设置为批量监听
        container.setBatchListener(true);
        return container;
    }



    // 监听器容器工厂(设置禁止KafkaListener自启动)
    @Bean
    public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
        Map<String, Object> propsMap =   consumerProps();

        container.setConsumerFactory(new DefaultKafkaConsumerFactory(propsMap));
        //禁止KafkaListener自启动
        container.setAutoStartup(false);
        return container;
    }







}
